/* Copyright (c) 2021 OceanBase and/or its affiliates. All rights reserved.
miniob is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:
         http://license.coscl.org.cn/MulanPSL2
THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details. */

//
// Created by Wangyunlai on 2022/12/07.
//

#pragma once

#include "sql/operator/physical_operator.h"
#include "common/log/log.h"
//小根堆排序算法
class OrderPhysicalOperator:public PhysicalOperator
{
public:
    OrderPhysicalOperator() = default;

    virtual ~OrderPhysicalOperator(){
        if(value_index_list_!=nullptr)delete[] value_index_list_;
        value_index_list_=nullptr;
        for(int i=0;i<order_value_lists_.size();i++){
            delete[] order_value_lists_[i];
            order_value_lists_[i]=nullptr;
        }
    }

  PhysicalOperatorType type() const override
    {
        return PhysicalOperatorType::ORDER;
    };

    RC open(Trx *trx) override
    {
        if (children_.empty())
        {
            return RC::SUCCESS;
        }
        if (is_open_)
        {
            return RC::SUCCESS;
        }
        if(order_types_.size()!=order_fields_.size()){
          LOG_WARN("Invalid argument:: order_types_ mismatch order_fields");
           return RC::INVALID_ARGUMENT;
        }
        PhysicalOperator::open(trx);
        PhysicalOperator *child = children_[0].get();
        RC rc = child->open(trx);
        if (rc != RC::SUCCESS)
        {
            LOG_WARN("failed to open child operator: %s", strrc(rc));
            return rc;
        }
        rc=init();
        if(rc!=RC::SUCCESS){
            LOG_WARN("failed to init order_physical_operator");
        }
        return RC::SUCCESS;
    }
    RC next() override;
   Tuple *current_tuple() override
    {
        return &val_tuple_;
    }
    RC set_order_fields(const std::vector<Field>& fields){
        order_fields_=fields;
        field_num_=fields.size();
        if(field_num_==0){
            LOG_WARN("Invalid argument :: none order field");
            return RC::INVALID_ARGUMENT;
        }
        return RC::SUCCESS;
    }
    RC set_order_types(const std::vector<OrderType>& order_types){
        order_types_=order_types;
        if(order_types_.size()==0){
            LOG_WARN("Invalid argument :: none order types");
            return RC::INVALID_ARGUMENT;
        }
        return RC::SUCCESS;
    }
private:
    int compare(const Value& left,const Value& right,const OrderType& order_type) const;   
    int compare(const Value* left_list,const Value* right_list) const;
    RC init();
    void sort();
    void swap(int p_index,int target_index);
    void adjustMinHeap(int index);
private:
    ValueListTuple val_tuple_;
    std::vector<std::vector<Value>> result_value_list_;
    std::vector<Field> order_fields_;
    std::vector<OrderType> order_types_;
    std::vector<const Value*> order_value_lists_;
    int *value_index_list_=nullptr;
    int field_num_=0;
    int row_num_=0;//行数，不为索引
};